package com.iot.service.mqtt.service.subscripe;

import com.iot.service.mqtt.model.entity.TlSubClient;
import com.iot.service.mqtt.service.StoreStrategy;
import com.iot.service.mqtt.service.session.ISessionService;
import com.iot.service.mqtt.service.topic.ITopicService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * @Author: hszhou
 * @Date: 2025/1/8 9:24
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@Service
@ConditionalOnProperty(prefix = "mqtt", name = "store-strategy", havingValue = StoreStrategy.REDIS)
@RequiredArgsConstructor
@Slf4j
public class RedisSubscriptionServiceImpl implements ISubscriptionService, ITopicService {


    private final ReactiveRedisTemplate<String, Object> reactiveRedisTemplate;

    private final ISessionService sessionService;

    private final String TOPIC_PREFIX = "topic:";


    @Override
    public Mono<Boolean> subscribe(TlSubClient clientSub) {

        return this.unsubscribe(clientSub)
                .flatMap(e -> reactiveRedisTemplate.opsForHash().put(TOPIC_PREFIX + clientSub.getTopic(), clientSub.getClientId(), clientSub.getQos())
                        .doOnSuccess(item -> {
                            // log.info("客户端【{}】订阅【{}】了,结果【{}】",clientSub.getClientId(),clientSub.getTopic(),e);
                        }).doOnError(error -> {
                            log.error("客户端【{}】订阅【{}】失败,原因【{}】", clientSub.getClientId(), clientSub.getTopic(), error.getMessage());
                        }));

    }

    @Override
    public Mono<Boolean> unsubscribe(String clientId, String topic) {
        TlSubClient subClient = new TlSubClient();
        subClient.setTopic(topic);
        subClient.setClientId(clientId);
        return this.unsubscribe(subClient);
    }

    public Mono<Boolean> unsubscribe(TlSubClient subClient) {

        return reactiveRedisTemplate.opsForHash().remove(TOPIC_PREFIX + subClient.getTopic(), subClient.getClientId())
                .map(value -> value > 0);
               // .doOnSuccess(e -> log.info("客户端【{}】取消订阅【{}】了,【{}】", subClient.getClientId(), subClient.getTopic(), e));
    }


    @Override
    public Flux<TlSubClient> find(String topic) {
        return reactiveRedisTemplate.keys(TOPIC_PREFIX + "*")
                .filter(key -> match(topic, key.substring(TOPIC_PREFIX.length())))
                .flatMap(key -> {
                    String actualTopic = key.substring(TOPIC_PREFIX.length());
                    return reactiveRedisTemplate.opsForHash().entries(key)
                            .map(entry -> {
                                String clientId = (String) entry.getKey();
                                int qos = (int) entry.getValue();
                                return new TlSubClient(qos, clientId, actualTopic);
                            });
                });
    }

    @Override
    public Mono<Boolean> clear(String clientId) {
       return sessionService.find(clientId)
                .flatMap(session -> Flux.fromIterable(session.getTopics())
                        .flatMap(topic -> this.unsubscribe(clientId, topic))
                        .then()
                        .thenReturn(true)
                        .onErrorReturn(false))
                .defaultIfEmpty(false);
    }
}
